/*
 * =====================================================================================
 *
 *       Filename:  redis_conn.cpp
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  03/15/2018 04:32:21 PM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  YOUR NAME (), 
 *   Organization:  
 *
 * =====================================================================================
 */
#include <iostream>
#include <algorithm>
#include <stdlib.h>
#include <set>
#include "redis_conn.h"
#include "redis_cluster.h"

using namespace std;
struct ClusterNode;

RedisConn::RedisConn(const string& ip, unsigned int port,
            const struct timeval& tmout):
    m_ctx(NULL),m_ip(ip),m_port(port),m_tmout(tmout)
{}

RedisConn::~RedisConn()
{
    redisFree(m_ctx);
    m_ctx = NULL;
}


bool RedisConn::connect()
{
    if (m_ctx != NULL)
    {
        redisFree(m_ctx);
        m_ctx = NULL;
    }
    
    if(m_tmout.tv_sec > 0 || m_tmout.tv_usec > 0)
    {
        m_ctx = redisConnectWithTimeout(m_ip.c_str(), m_port,m_tmout);
    }
    else
    {
        m_ctx = redisConnect(m_ip.c_str(), m_port);
    }

    if (m_ctx == NULL || m_ctx->err )
    {
        return false;
    }

    return true;
}

bool RedisConn::connected()
{
     return !(m_ctx == NULL || m_ctx->err != 0);
}

bool RedisConn::loadClusterSlots(const struct timeval& tmout)
{
    RedisCluster* ptr = RedisCluster::instance();
    if (ptr == NULL)
        return false;
	// 不能清除ptr->m_conns，重连时，会使用其中的可用连接
	ptr->m_nodes.clear();

	if (!connected() && !connect())
        return false;
    
	// 初始化cluster中的 m_nodes 和 m_conns
    // 本函数已是 cluster 的友元函数
    redisReply* reply = (redisReply*)redisCommand(m_ctx,"CLUSTER SLOTS");
    if( reply->type != REDIS_REPLY_ARRAY )
        return false;
    
    /*
     *1) 1) (integer) 0
         2) (integer) 5460
         3) 1) "127.0.0.1"
            2) (integer) 7000
            3) "cee027fbfbf985f36ae0e7140e27b4599331eda1"
         4) 1) "127.0.0.1"
            2) (integer) 7003
            3) "33b47046954ee7c2ed79a6f8fce17304beb234f8" 
     */
    
    int countCluster = reply->elements;
	set<ClusterNode> tmpconns; // 为了对比原来的，以便从原来的集合中删除
    for( int i = 0; i < countCluster ; ++i)
    {
        if( !(reply->element[i]->type== REDIS_REPLY_ARRAY &&
           reply->element[i]->element[0]->type == REDIS_REPLY_INTEGER &&    // slot num
           reply->element[i]->element[1]->type == REDIS_REPLY_INTEGER       // slot num
           ))
		{
			return false;
		}
		
		// get slave master;
        int countSM = reply->element[i]->elements;
		if (countSM < 3)	// no available slave/master
			return false;

        int slot_min = reply->element[i]->element[0]->integer;
        int slot_max = reply->element[i]->element[1]->integer;
        slot_min = std::min(slot_min,slot_max);
        slot_max = std::max(slot_min,slot_max);
        
        SlotRange sr(slot_min,slot_max);
		// 从第3条记录开始取相关点(master 及 slave)
		for (int j = 2; j < countSM; ++j)
		{
			if (reply->element[i]->element[j]->type != REDIS_REPLY_ARRAY ||
				reply->element[i]->element[j]->elements < 3 )
			{
				return false;
			}
            if ( !(reply->element[i]->element[j]->element[0]->type == REDIS_REPLY_STRING && // ip
                reply->element[i]->element[j]->element[1]->type == REDIS_REPLY_INTEGER && // port
                reply->element[i]->element[j]->element[2]->type == REDIS_REPLY_STRING // node id
            ))
			{
				// 已初始化的不删除
				return false;
			}

            string       ip      = reply->element[i]->element[j]->element[0]->str;
            unsigned int port    = reply->element[i]->element[j]->element[1]->integer;
            string       nodeid  = reply->element[i]->element[j]->element[2]->str;
            ClusterNode node(ip,port,nodeid);
            
			// ptr->m_nodes直接使用新值，原有的已清除
            ptr->m_nodes[sr].push_back(node);
			// ptr->m_conns要兼顾原有的，重新加载时有用
			tmpconns.insert(node);
			if (ptr->m_conns.find(node) == ptr->m_conns.end())
			{
				// 连接推到使用时
				ptr->m_conns[node] = new RedisConn(ip,port,tmout);
			}
        }
    }

	// 从结果中删除已不存在的RedisConn
	set<ClusterNode>::iterator itNow;
	map<ClusterNode,RedisConn*>::iterator itOld;
	for (itNow = tmpconns.begin(); itNow != tmpconns.end(); ++itNow)
	{
		itOld = ptr->m_conns.find(*itNow);
		if (itOld == ptr->m_conns.end())
		{
			delete itOld->second;
			ptr->m_conns.erase(itOld);
		}
	}
    return true;
}

bool RedisConn::exec(const string& cmd)
{
    if (!connected())
        connect();
    
    // 初始化成功 or 失败
    if (!connected())
        return false;
    
    redisReply* rsp = (redisReply*)redisCommand(m_ctx,cmd.c_str());
    if (rsp != NULL)
        freeReplyObject(rsp);

    return m_ctx->err == 0;
}

string RedisConn::to_lower(const string& str)
{
	if (str.empty())
		return str;
					    
	unsigned i = 0;
	string tmp(str.length(),'\0');
	while ( i < str.length())
	{   
		tmp[i] = char(tolower(str[i]));
		++i;
	}   
	return tmp;
}

int RedisConn::stringSplit(const string& str, const string& sep, vector<string> &vec )
{
	if (str.empty())
		return 0;

	size_t posBegin = 0;
	size_t posEnd = str.find(sep,posBegin);
	while(posEnd != string::npos)
	{   
		vec.push_back(str.substr(posBegin,posEnd-posBegin));
		posBegin = posEnd + sep.length();
		posEnd = str.find(sep,posBegin);
	}   
	if (posBegin < str.length() )
		vec.push_back(str.substr(posBegin));
	else if (posBegin >= str.length() )
		vec.push_back("");

	return vec.size();
}

RedisConn* RedisConn::getMovedConn(const string& errmsg)
{
	string msg = to_lower(errmsg);
	RedisConn* pconn = NULL;
	// when ASK or MOVED, find the correct connection
	// MOVED 10263 127.0.0.1:7001
	vector<string> ipport;
	stringSplit(msg," ",ipport);
	if (ipport.size() < 3)
		return NULL;

	RedisCluster* pCluster = RedisCluster::instance();
	if (pCluster == NULL)
		return false;

	string& ip = ipport[2];
	unsigned int port = atoi(ipport[1].c_str());
	
	if (msg.find("ask") != string::npos)
	{
		// asking不用重新加载集群配置
		// 变动后重新加载集群配置
		// pCluster->reloadSlots();
		pconn = pCluster->getConn(ip,port);
		if (pconn!=NULL)
		{
			pconn->exec("ASKING");
			cout<<"asking:"<<ip<<":"<<port<<endl;
		}
	}
	else if (msg.find("moved") != string::npos)
	{
		// 变动后重新加载集群配置
		pCluster->reloadSlots();
		pconn = pCluster->getConn(ip,port);
		cout<<"moved to:"<<ip<<":"<<port<<endl;
	}
		
	else if (errmsg.find("clusterdown") != string::npos)
	{
		// do nothing
	}

	return pconn;
}

bool RedisConn::getKey(const string& key,string& value)
{
    if (!connected())
        connect();
	
	// 初始化成功 or 失败
	bool bOk = false;
	redisReply* rsp = (redisReply*)redisCommand(m_ctx, "GET %b", key.c_str(),key.size());
	if (rsp != NULL)
	{   
		if (rsp->type == REDIS_REPLY_STRING)
		{
			value = rsp->str;
			bOk = true;
		}
		else if(rsp->type == REDIS_REPLY_ERROR)
		{
			string errmsg ( rsp->str, rsp->len );
			// 里面会重新加载最新集群结构
			// 并获取已迁移到的节点
			RedisConn* pconn = getMovedConn(errmsg);
			if (pconn == NULL)
			{
				freeReplyObject(rsp);
				return false;
			}
			
			freeReplyObject(rsp);
			return pconn->getKey(key,value);
		}
		freeReplyObject(rsp);
	}
	return bOk;
}

bool RedisConn::setKeyValue(const string& key, const string &value, const string &ttl)
{
	if (!connected())
        connect();
	
	// 初始化成功 or 失败
	redisReply* rsp = (redisReply*)redisCommand(m_ctx, "SET %b %b EX %s",
		key.c_str(), key.size(), value.c_str(), value.size(), ttl.c_str());
	// m_ctx->err 检测连接中断 或者 server 挂掉
	if (m_ctx->err != 0)
	{
		if(rsp->type == REDIS_REPLY_ERROR)
		{
			string errmsg ( rsp->str, rsp->len );
			// 里面会重新加载最新集群结构
			// 并获取已迁移到的节点
			RedisConn* pconn = getMovedConn(errmsg);
			if (pconn == NULL)
			{
				freeReplyObject(rsp);
				return false;
			}
			
			freeReplyObject(rsp);
			return pconn->setKeyValue(key,value,ttl);
		}
	}
	
	if (rsp != NULL)
		freeReplyObject(rsp);
	
	return true;
}



